package com.group2.edu.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Set;

/**
 * @author 高耀
 * @date 2024/12/14 10:29<p></p>
 */
public class HBaseUtil {
    /**
     * 获取HBase连接的方法
     */
    public static Connection getConnection() {
        Configuration conf = new Configuration();
        //配置HBase地址
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104,hadoop105,hadoop106");
        try {
            //通过工厂创建HBase连接对象
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            //创建连接失败
            throw new RuntimeException(e);
        }
    }

    /**
     * 关闭连接的方法
     */
    public static void closeConnection(Connection connection) {
        try {
            if (connection != null && !connection.isClosed()) {//不为空且未关闭
                connection.close();
            }
        } catch (IOException e) {
            //关闭失败
            throw new RuntimeException(e);
        }
    }


    /**
     * 在HBase中创建表的方法
     *
     * @param connect   传入的HBase连接
     * @param namespace 命名空间，库
     * @param tableName 表名
     * @param families  列族，可以有多个
     */
    public static void createTable(
            Connection connect,
            String namespace,
            String tableName,
            String... families
    ) {
        //先判断列族是否合理
        if (families.length == 0) {
            System.out.println("列族至少需要一个!!!");
            return;
        }
        try (Admin admin = connect.getAdmin()) {//根据连接获取Admin对象,使用try的语法，可以自动关闭创建的admin
            //使用 命名空间 和 表名 创建表名对象
            TableName tableNameObj = TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(tableName));
            //判断要创建的表是否已经存在
            if (admin.tableExists(tableNameObj)) {
                System.out.println("要创建的" + namespace + "命名空间下的" + tableName + "表已经存在!!!");
                return;
            }
            //使用 表名对象 创建表的描述器Builder
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            //遍历列族数组
            for (String family : families) {
                //创建列族描述器
                ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder
                        .newBuilder(Bytes.toBytes(family))
                        .build();
                //设置列族
                tableDescriptorBuilder.setColumnFamily(familyDescriptor);
            }
            //通过描述器建表
            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("创建" + namespace + "命名空间下的" + tableName + "表成功!!!");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 删除HBase中的表
     *
     * @param connect   HBase连接
     * @param namespace 命名空间
     * @param tableName 表名
     */
    public static void deleteTable(Connection connect, String namespace, String tableName) {

        try (Admin admin = connect.getAdmin()) {//获取admin对象
            //创建TableName对象
            TableName tableNameObj = TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(tableName));
            //判断该表是否不存在
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("要删除的" + namespace + "命名空间下的" + tableName + "表不存在!!!");
                return;
            }
            //存在
            //先让该表失效
            admin.disableTable(tableNameObj);
            //再删除
            admin.deleteTable(tableNameObj);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    public static void deleteData(
            Connection connection,
            String namespace,
            String sinkTable,
            String rowKey
    ) {
        //创建表名对象
        TableName tableName =
                TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(sinkTable));
        try (Table table = connection.getTable(tableName)) { //根据表名对象获取Table对象
            //根据sinkRow创建Delete对象
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            //执行删除
            table.delete(delete);
            System.out.println(namespace + "命名空间下" + sinkTable + "表的" + rowKey + "行已被删除~~~");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void insertData(
            Connection connection,
            String namespace,
            String sinkTable,
            String rowKey,
            String sinkFamily,
            JSONObject dataJson
    ) {
        //创建表名对象
        TableName tableName =
                TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(sinkTable));
        try (Table table = connection.getTable(tableName)) { //根据TableName创建Table
            //根据rowKey创建Put对象
            Put put = new Put(Bytes.toBytes(rowKey));
            //获取数据所有字段名
            Set<String> columnSet = dataJson.keySet();
            //遍历
            for (String columnName : columnSet) {
                //根据字段名获取字段值
                String columnValue = dataJson.getString(columnName);
                if (columnValue != null && !columnValue.isEmpty()) {
                    //在put对象中添加列名和列值
                    //要列族(这里只有一个，不作特殊处理，直接放入)、列名、列值
                    put.addColumn(
                            Bytes.toBytes(sinkFamily),
                            Bytes.toBytes(columnName),
                            Bytes.toBytes(columnValue)
                    );
                }
            }
            //执行插入操作
            table.put(put);
            System.out.println("向" + namespace + "命名空间的" + sinkTable + "表的插入" + rowKey + "数据成功");
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    /**
     * 该方法用于到HBase中查询对应命名空间下对应表下指定rowKey值的数据封装成指定类型对象返回
     * @param connection HBase连接
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey rowKey
     * @param clz 封装类型
     * @param isConvertToLowerCamel 是否将Hbase中的蛇形命名法字段转成小驼峰命名法
     */
    public static <T>T getRow(
            Connection connection,
            String namespace,
            String tableName,
            String rowKey,
            Class<T> clz,
            boolean... isConvertToLowerCamel
    ) {
        //查看是否要蛇型转驼峰
        boolean isConvert = false;
        if (isConvertToLowerCamel.length > 0 && isConvertToLowerCamel[0] == true) {
            isConvert = true;
        }
        //根据命名空间和表名创建表名对象
        TableName tableNameObj
                = TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(tableName));
        //通过连接获取表对象
        try (Table table = connection.getTable(tableNameObj)){
            //根据rowKey创建Get对象
            Get get = new Get(Bytes.toBytes(rowKey));
            //根据Get对象从表中查询出结果
            Result result = table.get(get);
            //根据查询结果Result获取这一行所有单元格组成的数组
            List<Cell> cells = result.listCells();
            //如果查到了数据，准备封装
            if (cells != null && !cells.isEmpty()) {
                //使用反射创建封装对象
                T obj = clz.newInstance();
                //遍历单元格给对象赋值
                for (Cell cell : cells) {
                    //获取单元格的字段名
                    String fieldName = new String(CellUtil.cloneQualifier(cell));
                    //判断是否要转成小驼峰
                    if (isConvert) {
                        fieldName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL,fieldName);
                    }
                    //获取单元格的字段值
                    String fieldValue = new String(CellUtil.cloneValue(cell));
                    //给对象赋值
                    BeanUtils.setProperty(obj, fieldName, fieldValue);
                }
                //返回封装好的对象
                return obj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        //运行到这里说明没查到
        return null;
    }


    /**
     * 获取异步操作类型的HBase连接
     * @return
     */
    public static AsyncConnection getAsyncConnection() {
        Configuration conf = new Configuration();
        //配置HBase地址
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        try {
            //通过工厂创建HBase连接对象
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            //创建连接失败
            throw new RuntimeException(e);
        }
    }


    /**
     * 关闭异步连接的方法
     */
    public static void closeAsyncConnection(AsyncConnection connection) {
        try {
            if (connection != null && !connection.isClosed()) {//不为空且未关闭
                connection.close();
            }
        } catch (IOException e) {
            //关闭失败
            throw new RuntimeException(e);
        }
    }


    /**
     * 以异步的方式到HBase表中查询维度表的数据
     * @param conn 连接
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey rowKey
     * @return 直接封装成JSONObject形式返回， 属性命名也不用变化
     */
    public static JSONObject getAsyncDimRow(AsyncConnection conn, String namespace, String tableName, String rowKey) {
        //创建TableName对象
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        //获取Table对象
        AsyncTable<AdvancedScanResultConsumer> table = conn.getTable(tableNameObj);
        //创建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            //在table中查询
            Result result = table.get(get).get();
            //获取查询结果一行的单元格组成的集合
            List<Cell> cells = result.listCells();
            if (cells != null && !cells.isEmpty()) {
                //创建要封装的对象
                JSONObject jsonObject = new JSONObject();
                for (Cell cell : cells) {
                    //取出每个单元格的名字和值赋给要返回的对象
                    String fieldName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String fieldValue = Bytes.toString(CellUtil.cloneValue(cell));
                    BeanUtils.setProperty(jsonObject, fieldName, fieldValue);
                }
                return jsonObject;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    public static JSONObject readDimAsync(AsyncConnection hbaseAsyncConn,String namespace,String tableName,String rowkey){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = hbaseAsyncConn.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowkey));
        try {
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject dimJsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    dimJsonObj.put(columnName,columnValue);
                }
                return dimJsonObj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }
}